package com.example.kafkaconsumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * @author: wukangrong
 * @date: 2021/1/29 16:50
 * @description:
 */
@Slf4j
@Component
public class Consumer {

    @KafkaListener(topics = {"${kafka.custom.topic.name}"}, concurrency = "2")
    public void processMsg(ConsumerRecord consumerRecord, Acknowledgment ack) throws InterruptedException {
        if (Objects.nonNull(consumerRecord) && Objects.nonNull(ack)) {
            log.info("接收到信息:{}, offset:{}", consumerRecord.value(), consumerRecord.offset());
            TimeUnit.SECONDS.sleep(2);
            ack.acknowledge();
            log.info("提交成功 offset:{}", consumerRecord.offset());
        }
    }


}
